package com.zp.qualitypractice.rx;

import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Consumer;
import io.reactivex.processors.PublishProcessor;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.Subject;

/**
 * Created by izp on 2017/8/14.
 */

public class InnerRxBus {
    private static volatile InnerRxBus mDefaultInstance;
    private final Subject<Object> _bus;
    private InnerRxBus() {
        //调用toSerialized()方法，保证线程安全
        _bus = PublishSubject.create().toSerialized();
    }
    public static InnerRxBus getDefault() {
        if (mDefaultInstance == null) {
            synchronized (InnerRxBus.class) {
                if (mDefaultInstance == null) {
                    mDefaultInstance = new InnerRxBus();
                }
            }
        }
        return mDefaultInstance;
    }

    public void send(String tag, Object object) {
        EventObject eventObject = new EventObject(tag, object);
        _bus.onNext(eventObject);
    }
    public interface ReceiveOnUiThread {
        void OnReceive(String tag, Object object);
    }
    public interface ReceiveOnIOThread {
        void OnReceive(String tag, Object object);
    }
    public void registerOnUiThread(final ReceiveOnUiThread receiveOnUiThread) {
//        _bus.observeOn(AndroidSchedulers.mainThread())
//                .subscribe(new Action1<EventObject>() {
//                    @Override
//                    public void call(EventObject eventObject) {
//                        receiveOnUiThread.OnReceive(eventObject.getTag(), eventObject.getObject());
//                    }
//                });
        _bus.observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Object>() {
                    @Override
                    public void accept(@NonNull Object o) throws Exception {
                        if (o instanceof EventObject){
                            EventObject eventObject = (EventObject) o;
                            receiveOnUiThread.OnReceive(eventObject.getTag(), eventObject.getObject());
                        }
                    }
                });
    }
    public void registerOnIOThread(final ReceiveOnIOThread receiveOnIOThread) {
//        _bus.observeOn(Schedulers.io())
//                .subscribe(new Action1<EventObject>() {
//                    @Override
//                    public void call(EventObject eventObject) {
//                        receiveOnIOThread.OnReceive(eventObject.getTag(), eventObject.getObject());
//                    }
//                });
        _bus.observeOn(Schedulers.io())
                .subscribe(new Consumer<Object>() {
                    @Override
                    public void accept(@NonNull Object o) throws Exception {
                        if (o instanceof EventObject){
                            EventObject eventObject = (EventObject) o;
                            receiveOnIOThread.OnReceive(eventObject.getTag(), eventObject.getObject());
                        }
                    }
                });
    }
    private static class EventObject {
        private String tag;
        private Object object;
        public EventObject(String tag, Object object) {
            this.tag = tag;
            this.object = object;
        }
        public String getTag() {
            return tag;
        }
        public Object getObject() {
            return object;
        }
    }
}
